package cn.doitedu.etl;

import cn.doitedu.udfs.GeoHashUDF;
import cn.doitedu.udfs.Map2JsonUDF;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/6
 * @Desc: 学大数据，上多易教育
 * 用户行为日志ods到dwd加工： 主要工作是公共维度退维
 * 数据源： kafka的topic： ods-user-action-log
 * <p>
 * 目的地： kafka的topic： dwd-user-action-log
 * doris中的明细表：dwd.user_action_detail
 **/
public class Job01_UserActionLogDimTui {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///D:/flink/ck");
        env.disableOperatorChaining();

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中的数据源topic
        tenv.executeSql(
                " create table ods_events_kafka(                               " +  // 物理字段
                        "   release_channel string,                            " +  // 物理字段
                        "   device_type string,                                " +  // 物理字段
                        "   session_id string,                                 " +  // 物理字段
                        "   lat double,                                        " +  // 物理字段
                        "   lng double,                                        " +  // 物理字段
                        "   user_name string,                                  " +  // 物理字段
                        "   event_id string,                                   " +  // 物理字段
                        "   action_time bigint,                                " +  // 物理字段
                        "   properties map<string,string>,                     " +  // 物理字段
                        "   pt as proctime()                                   " +  // 表达式字段
                        " ) WITH (                                             " +
                        "   'connector' = 'kafka',                             " +
                        //"   'topic' = 'ods-user-action-log',                   " +
                        "   'topic' = 'backpressure',                   " +
                        "   'properties.bootstrap.servers' = 'doitedu:9092',   " +
                        "   'properties.group.id' = 'doit47_g1',               " +
                        "   'scan.startup.mode' = 'latest-offset',             " +
                        "   'value.format' = 'json',                           " +
                        "   'value.fields-include' = 'EXCEPT_KEY'              " +
                        " )                                                    "
        );


        // 建表，映射kafka中的目标topic
        tenv.executeSql(
                "create table dwd_events_kafka(     " +
                        "     release_channel    string             " +
                        "    ,device_type        string             " +
                        "    ,session_id         string             " +
                        "    ,lat                double             " +
                        "    ,lng                double             " +
                        "    ,user_name          string             " +
                        "    ,event_id           string             " +
                        "    ,action_time        bigint             " +
                        "    ,properties         map<string,string> " +

                        "    ,user_id bigint             " +
                        "    ,member_level_id bigint    " +
                        "    ,nickname string           " +
                        "    ,phone string              " +
                        "    ,status int                " +
                        "    ,create_time timestamp(3)  " +
                        "    ,icon string               " +
                        "    ,gender int                " +
                        "    ,birthday date             " +
                        "    ,register_city string      " +
                        "    ,job string                " +
                        "    ,personalized_signature string " +
                        "    ,source_type int           " +
                        "    ,integration int           " +
                        "    ,growth int                " +
                        "    ,luckey_count int          " +
                        "    ,history_integration int   " +
                        "    ,modify_time timestamp(3)  " +

                        "    ,province   string        " +
                        "    ,city       string        " +
                        "    ,region     string        " +

                        "    ,page_type       string   " +
                        "    ,page_service    string   " +
                        "    ,page_lanmu      string   " +
                        "    ,page_pindao     string   " +
                        ") WITH (                      " +
                        "    'connector' = 'kafka',    " +
                        "    'topic' = 'dwd-user-action-log',  " +
                        "    'properties.bootstrap.servers' = 'doitedu:9092',  " +
                        "    'properties.group.id' = 'doit44_g1',              " +
                        "    'scan.startup.mode' = 'latest-offset',            " +
                        "    'value.format' = 'json',                          " +
                        "    'value.fields-include' = 'EXCEPT_KEY'             " +
                        ")                                                     ");


        // 建表，映射doris中的目标表
        tenv.executeSql(
                " create table user_events_detail_doris(                            " +
                        "     event_id                   string                     " +
                        "     ,province                  string                     " +
                        "     ,city                      string                     " +
                        "     ,region                    string                     " +
                        "     ,dt                        date                       " +
                        "     ,release_channel           string                     " +
                        "     ,device_type               string                     " +
                        "     ,session_id                string                     " +
                        "     ,lat                       double                     " +
                        "     ,lng                       double                     " +
                        "     ,user_name                 string                     " +
                        "     ,action_time               bigint                     " +
                        "     ,properties                string                     " +
                        "     ,user_id                   bigint                     " +
                        "     ,member_level_id           int                        " +
                        "     ,nickname                  string                     " +
                        "     ,phone                     string                     " +
                        "     ,status                    int                        " +
                        "     ,create_time               timestamp(3)               " +
                        "     ,icon                      string                     " +
                        "     ,gender                    int                        " +
                        "     ,birthday                  date                       " +
                        "     ,register_city             string                     " +
                        "     ,job                       string                     " +
                        "     ,personalized_signature    string                     " +
                        "     ,source_type               int                        " +
                        "     ,integration               int                        " +
                        "     ,growth                    int                        " +
                        "     ,luckey_count              int                        " +
                        "     ,history_integration       int                        " +
                        "     ,modify_time               timestamp(3)               " +
                        "     ,page_type                 string                     " +
                        "     ,page_service              string                     " +
                        "     ,page_lanmu                string                     " +
                        "     ,page_pindao               string                     " +
                        " ) WITH (                                                  " +
                        "    'connector' = 'doris',                                 " +
                        "    'fenodes' = 'doitedu:8030',                            " +
                        "    'table.identifier' = 'dwd.user_events_detail',         " +
                        "    'username' = 'root',                                   " +
                        "    'password' = 'root',                                   " +
                        "    'sink.label-prefix' = 'doris_label-0x6"+System.currentTimeMillis()+"'" +
                        " )                                                         "

        );


        // 创建3张维表的逻辑映射表
        tenv.executeSql(
                        " create table ums_member_hbase(          " +
                        "   username string                       " +
                        "   ,f ROW<                               " +
                        "    id bigint                             " +
                        "   ,member_level_id bigint               " +
                        "   ,nickname string                      " +
                        "   ,phone string                         " +
                        "   ,status int                           " +
                        "   ,create_time timestamp(3)             " +
                        "   ,icon string                          " +
                        "   ,gender int                           " +
                        "   ,birthday date                        " +
                        "   ,city string                          " +
                        "   ,job string                           " +
                        "   ,personalized_signature string        " +
                        "   ,source_type int                      " +
                        "   ,integration int                      " +
                        "   ,growth int                           " +
                        "   ,luckey_count int                     " +
                        "   ,history_integration int              " +
                        "   ,modify_time timestamp(3)   >         " +
                        " ) WITH (                                " +
                        "  'connector' = 'hbase-2.2',             " +
                        "  'table-name' = 'dim_user_info',        " +
                        "  'lookup.async' = 'true',               " +
                        "  'zookeeper.quorum' = 'doitedu:2181'    " +
                        " )                                        "
        );

        tenv.executeSql(
                " create table dim_geo_info_hbase (                  " +
                        "   geohash string                        " +
                        "   ,f ROW<                               " +
                        "    province string                      " +
                        "   ,city string                          " +
                        "   ,region string   >                    " +
                        " ) WITH (                                " +
                        "  'connector' = 'hbase-2.2',             " +
                        "  'table-name' = 'dim_gps_info',         " +
                        "  'zookeeper.quorum' = 'doitedu:2181',   " +
                        "  'lookup.cache' = 'PARTIAL',                          " +
                        "  'lookup.partial-cache.max-rows' = '100000',          " +
                        "  'lookup.partial-cache.expire-after-access' = '1 h',  " +
                        "  'lookup.partial-cache.cache-missing-key' = 'true'    " +
                        " )                                        "
        );


        tenv.executeSql(
                " create table dim_page_info_hbase (              " +
                        "   url_prefix string                     " +
                        "   ,f ROW<                               " +
                        "     page_lanmu string                    " +
                        "    ,page_pindao string                  " +
                        "    ,page_service string                 " +
                        "    ,page_type string                    " +
                        "    >                                    " +
                        " ) WITH (                                " +
                        "  'connector' = 'hbase-2.2',             " +
                        "  'table-name' = 'dim_page_info',        " +
                        "  'zookeeper.quorum' = 'doitedu:2181',    " +
                        "  'lookup.cache' = 'PARTIAL',                          " +
                        "  'lookup.partial-cache.max-rows' = '100000',          " +
                        "  'lookup.partial-cache.expire-after-access' = '1 h',  " +
                        "  'lookup.partial-cache.cache-missing-key' = 'true'    " +
                        " )                                                     "
        );


        // 维表关联
        tenv.createTemporaryFunction("gps_geo", GeoHashUDF.class);
        tenv.executeSql(
                "create temporary view joined_view as " +
                "SELECT                   \n" +
                "  t1.release_channel     \n" +
                " ,t1.device_type         \n" +
                " ,t1.session_id          \n" +
                " ,t1.lat                 \n" +
                " ,t1.lng                 \n" +
                " ,t1.user_name           \n" +
                " ,t1.event_id            \n" +
                " ,t1.action_time         \n" +
                " ,t1.properties          \n" +
                "                         \n" +
                " ,t2.f.id as user_id     \n" +
                " ,t2.f.member_level_id     \n" +
                " ,t2.f.nickname            \n" +
                " ,t2.f.phone               \n" +
                " ,t2.f.status              \n" +
                " ,t2.f.create_time         \n" +
                " ,t2.f.icon                \n" +
                " ,t2.f.gender              \n" +
                " ,t2.f.birthday            \n" +
                " ,t2.f.city  as register_city      \n" +
                " ,t2.f.job                       \n" +
                " ,t2.f.personalized_signature    \n" +
                " ,t2.f.source_type               \n" +
                " ,t2.f.integration               \n" +
                " ,t2.f.growth                    \n" +
                " ,t2.f.luckey_count              \n" +
                " ,t2.f.history_integration       \n" +
                " ,t2.f.modify_time               \n" +
                "                             \n" +
                " ,t3.f.province         \n" +
                " ,t3.f.city             \n" +
                " ,t3.f.region           \n" +
                "                             \n" +
                " ,t4.f.page_type       \n" +
                " ,t4.f.page_service    \n" +
                " ,t4.f.page_lanmu      \n" +
                " ,t4.f.page_pindao     \n" +
                "                        \n" +
                "from ods_events_kafka t1 \n" +
                "left join ums_member_hbase    for system_time as of t1.pt as  t2 on t1.user_name = t2.username\n" +
                "left join dim_geo_info_hbase  for system_time as of t1.pt as  t3 on   reverse(gps_geo(t1.lat,t1.lng)) = t3.geohash\n" +
                "left join dim_page_info_hbase for system_time as of t1.pt as  t4 on  regexp_extract(t1.properties['url'],'(/.*/).*?',0)= t4.url_prefix \n" +
                "\n");


        // 结果写入kafka
        tenv.executeSql("insert into dwd_events_kafka select * from joined_view");


        // 结果写入doris
        tenv.createTemporaryFunction("map_json", Map2JsonUDF.class);
        tenv.executeSql(
                " insert into user_events_detail_doris                  " +
                        " select                                                " +
                        "     event_id                                          " +
                        "     ,province                                         " +
                        "     ,city                                             " +
                        "     ,region                                           " +
                        "     ,to_date(date_format(to_timestamp_ltz(action_time,3),'yyyy-MM-dd')) as dt   " +
                        "     ,release_channel                                  " +
                        "     ,device_type                                      " +
                        "     ,session_id                                       " +
                        "     ,lat                                              " +
                        "     ,lng                                              " +
                        "     ,user_name                                        " +
                        "     ,action_time                                      " +
                        "     ,map_json(properties) as properties               " +
                        "     ,user_id                                          " +
                        "     ,cast(member_level_id as int) as member_level_id  " +
                        "     ,nickname                                         " +
                        "     ,phone                                            " +
                        "     ,status                                           " +
                        "     ,create_time                                      " +
                        "     ,icon                                             " +
                        "     ,gender                                           " +
                        "     ,birthday                                         " +
                        "     ,register_city                                    " +
                        "     ,job                                              " +
                        "     ,personalized_signature                           " +
                        "     ,source_type                                      " +
                        "     ,integration                                      " +
                        "     ,growth                                           " +
                        "     ,luckey_count                                     " +
                        "     ,history_integration                              " +
                        "     ,modify_time                                      " +
                        "     ,page_type                                        " +
                        "     ,page_service                                     " +
                        "     ,page_lanmu                                       " +
                        "     ,page_pindao                                      " +
                        "                                                       " +
                        " from joined_view                                      "
        );


    }

}
